package com.example.demo.hutool.redis.stream;

import cn.hutool.core.map.MapUtil;
import redis.clients.jedis.StreamEntry;
import redis.clients.jedis.StreamEntryID;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class RedisStreamTest {

//xadd:添加消息
//xread:消费消息
//xdel:删除消息
//xgroup:消费组管理
//xreadgroup:分组消费
//xrange: 获取没有删除标识的消息列表
//xlen:获取整个Stream的消息长度
//del：删除整个Stream的消息
//xdel:删除消息
//xpending: 查看未处理消息
//xack:确认消息已经被处理
//xclaim:转移消息
//xinfo:查看队列信息
//xtrim:消息队列容量
//xrevrange:逆序获取消息队列中的消息

    private static IStreamServer iStreamServer = new StreamServiceImpl();


    public void setGroup() throws Exception {
        //创建 流名为 video 分组为 group1
        String res = iStreamServer.xgroupCreate("video", "group1", true);
        //创建 流名为 audio分组为 group1
        String res2 = iStreamServer.xgroupCreate("audio", "group1", true);


    }

    public static void main(String[] args) throws Exception {
        RedisStreamTest redisStreamTest = new RedisStreamTest();

        String res = iStreamServer.xgroupCreate("video", "group3", false);
        //创建 流名为 audio分组为 group1
        String res2 = iStreamServer.xgroupCreate("audio", "group3", true);
       redisStreamTest.xadd();

//        redisStreamTest.xreadGroup();
    }

    public void xadd() throws Exception {
        Long time = System.currentTimeMillis();
        StreamEntryID streamEntryID = new StreamEntryID(String.format("%s-%s", time, 1));
        Map<String, String> id1 = new HashMap<>();
        id1.put("mmm", 1 + "");
         iStreamServer.xadd("audio", streamEntryID, id1);
         iStreamServer.xadd("video", streamEntryID, id1);
        //向audio中插入100条消息
//        for (int i = 1; i < 2; i++) {
//            StreamEntryID streamEntryID = new StreamEntryID(String.format("%s-%s", time, i));
//            Map<String, String> id1 = new HashMap<>();
//            id1.put("id", i + "");
//            StreamEntryID id = iStreamServer.xadd("audio", streamEntryID, id1);
//            System.out.println(id);
//        }
//
//        //向video中插入100条消息
//        for (int i = 1; i < 2; i++) {
//            StreamEntryID streamEntryID = new StreamEntryID(String.format("%s-%s", time, i));
//            Map<String, String> id1 = new HashMap<>();
//            id1.put("id", i + "");
//            StreamEntryID id = iStreamServer.xadd("video", streamEntryID, id1);
//            System.out.println(id);
//        }
    }


    public void xreadGroup() throws Exception {

//        Map<String, StreamEntryID> t = MapUtil.of("video", null);
//       // Map<String, StreamEntryID> t2 = MapUtil.of("audio", null);
//
//        Map.Entry<String, StreamEntryID> video = t.entrySet().stream().findFirst().get();
//     //   Map.Entry<String, StreamEntryID> audio = t2.entrySet().stream().findFirst().get();
//
//        //client1 用 group1 分组 从 video/audio 流中获取两个消息 ,同一分组中不同客户端可共享消费消息(client1消费一条，client2消息偏移量增加1 )，不同分组中的客户端消息不影响（client1消费一条，不影响client2消息偏移量）
//        List<Map.Entry<String, List<StreamEntry>>> list = iStreamServer.xreadGroup("group3", "client4", 10, video);
//
//        if (list != null && list.size() > 0) {
//            list.forEach(x -> {
//                System.out.println(x.getKey() + "-->" + x.getValue());
//            });
//        }
        ArrayList<String> limit = new ArrayList<>();
        limit.add("video");
        limit.add("audio");
        Map.Entry<String, StreamEntryID>[] ps =  new Map.Entry[2];
        for (int i = 0; i < limit.size(); i++) {
            Map<String, StreamEntryID> map = new HashMap<>();
            map.put(limit.get(i), null);
            Map.Entry<String, StreamEntryID> video = map.entrySet().stream().findFirst().get();
            System.out.println();
            ps[i] = video;

        }


        //client1 用 group1 分组 从 video/audio 流中获取两个消息 ,同一分组中不同客户端可共享消费消息(client1消费一条，client2消息偏移量增加1 )，不同分组中的客户端消息不影响（client1消费一条，不影响client2消息偏移量）

        List<Map.Entry<String, List<StreamEntry>>> list = iStreamServer.xreadGroup("group3", "client", 10000,  ps);

        if (list != null && list.size() > 0) {
            list.forEach(x -> {
                System.out.println(x.getKey() + "-->" + x.getValue());
            });
        }
        //输出
        /**
         video-->[1610607445337-2 {id=2}, 1610607445337-3 {id=3}]
         audio-->[1610608368683-2 {id=2}, 1610608368683-3 {id=3}]
         **/


    }


    public void xrange() throws Exception {
        //video中按正序 从1610607445337-10 到1610607445337-20 中获取3条消息
        List<StreamEntry> streamEntries = iStreamServer.xrange("video", new StreamEntryID("1610607445337-10"), new StreamEntryID("1610607445337-20"), 3);
        streamEntries.forEach(x -> System.out.println(x));
        //输出
        /**
         1610607445337-10 {id=10}
         1610607445337-11 {id=11}
         1610607445337-12 {id=12}
         **/
    }


    public void xrevrange() throws Exception {
        //video中按倒序 从1610607445337-80 到1610607445337-0 中获取3条消息
        List<StreamEntry> streamEntries = iStreamServer.xrevrange("video", new StreamEntryID("1610607445337-80"), new StreamEntryID("1610607445337-0"), 3);
        streamEntries.forEach(x -> System.out.println(x));
        //输出
        /**
         1610607445337-80 {id=80}
         1610607445337-79 {id=79}
         1610607445337-78 {id=78}
         **/
    }

    public void xread() throws Exception {
        //从1610607445337-2 开始获取
        Map<String, StreamEntryID> t = MapUtil.of("video", new StreamEntryID("1610607445337-2"));
        Map<String, StreamEntryID> t2 = MapUtil.of("audio", null);
        Map.Entry<String, StreamEntryID> video = t.entrySet().stream().findFirst().get();
        Map.Entry<String, StreamEntryID> audio = t2.entrySet().stream().findFirst().get();
        // 从video 1610607445337-2 开始，从audio 0开始获取 2条消息
        List<Map.Entry<String, List<StreamEntry>>> list = iStreamServer.xread(2, video, audio);
        list.forEach(x -> {
            System.out.println(x.getKey() + "-->" + x.getValue());
        });
        //输出
        /**
         audio-->[1610608368683-0 {id=0}, 1610608368683-1 {id=1}]
         video-->[1610607445337-3 {id=3}, 1610607445337-4 {id=4}]
         **/
    }

}
